package cn.liuwenlin.mq;

import cn.liuwenlin.mq.entity.Girl;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ScheduledMessage;

import javax.jms.*;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

public class QueueSender3 {

    public static void main(String[] args) {
        //1.建立连接工厂
//        ActiveMQConnectionFactory activeMQConnectionFactory
//                = new ActiveMQConnectionFactory(
//                        ActiveMQConnectionFactory.DEFAULT_USER,
//                        ActiveMQConnectionFactory.DEFAULT_PASSWORD,
//                "tcp://master.tony.com:61616");
        //使用nio配置的连接传输协议,提高数据传输效率
        ActiveMQConnectionFactory activeMQConnectionFactory
                = new ActiveMQConnectionFactory(
                ActiveMQConnectionFactory.DEFAULT_USER,
                ActiveMQConnectionFactory.DEFAULT_PASSWORD,
                "nio://hadoop.master.com:5671");
        Connection connection = null;
        try {
            //2.获取一个向Active MQ的连接
            connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //3.获取session（会话可以设置是否开启事务，如果开启，则需要在发送消息后做提交操作，否则消息不会投递）
//            Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            //4.找目的地，获取destination，消费端，也会从这个目的地取消息
            Queue mq = session.createQueue("activemq");

            //5.1消息生产者
            MessageProducer producer = session.createProducer(mq);
            String uuid = UUID.randomUUID().toString();
            /**
             * 注:不管是创建TemporaryQueue(独立于不同的session)
             * 还是单独创建一个新的用于生产端和消费端的消息ack确认机制
             * (即消息的JMSReplyTo的设定和获取)
             */
//            TemporaryQueue TemporaryQueue = session.createTemporaryQueue();
            //5.2创建消息
            TextMessage msg = null;
            for(int i = 0; i < 10; i++){
                msg = session.createTextMessage("Hi activemq" + i);
                msg.setJMSCorrelationID(uuid);
                msg.setStringProperty("selector", "pub");
                //5.3向目的地写入消息
                producer.send(msg);
                System.out.println(msg + " has been send...");

//                if(i % 4 == 0){
//                    session.rollback();
//                } else {
//                    session.commit();
//                }

                TimeUnit.MILLISECONDS.sleep(10);
            }

            MessageConsumer consumer = session.createConsumer(mq, "JMSCorrelationID='" + uuid + "' and selector='recv'");
            TemporaryQueue temporaryQueue = session.createTemporaryQueue();
            consumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    TextMessage msg = (TextMessage)message;
                    System.out.println("---Get back Message---");
                    try {
                        System.out.println("receive msg:" + msg.getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });

            while(true){
                TimeUnit.MILLISECONDS.sleep(10);
            }

        } catch (JMSException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {

            //6.关闭连接
            if(connection != null){
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }

            System.out.println("exit...");
        }

    }

}
